package com.lagou.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
//import org.junit.After;
//import org.junit.Before;
//import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class MyHbaseClient {
    private Configuration conf = null;
    private Connection conn = null;
    private HBaseAdmin admin = null;
////    @Before
//    public void init() throws IOException {
//        conf = HBaseConfiguration.create();
//        conf.set("hbase.zookeeper.quorum", "master");
//        conf.set("hbase.zookeeper.property.clientPort", "2181");
//        conn = ConnectionFactory.createConnection(conf);
//    }
//
////    @After

    public MyHbaseClient() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conn = ConnectionFactory.createConnection(conf);
        admin = (HBaseAdmin) conn.getAdmin();
    }

//    @Test
    public void createTable() throws IOException {
        HTableDescriptor friends  = new HTableDescriptor(TableName.valueOf("friends"));
        friends.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(friends);
        System.out.println("friends table has benn created !");
    }

    public boolean checkTableExists() throws IOException {
        return admin.tableExists(TableName.valueOf("friends"));
    }

//    @Test
    public void putData() throws IOException {
        Table friends = conn.getTable(TableName.valueOf("friends"));
        friends.setOperationTimeout(1000);
        friends.setRpcTimeout(1000);
        System.out.println("get the connection of the table");
        // 创建5个uid
        Put put1 = new Put(Bytes.toBytes("uid1"));
        Put put2 = new Put(Bytes.toBytes("uid2"));
        Put put3 = new Put(Bytes.toBytes("uid3"));
        Put put4 = new Put(Bytes.toBytes("uid4"));
        Put put5 = new Put(Bytes.toBytes("uid5"));
        // 将对应关系放入相应的列
        put1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        put1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid5"), Bytes.toBytes("uid5"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid3"), Bytes.toBytes("uid3"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid2"), Bytes.toBytes("uid2"));
        put4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid5"), Bytes.toBytes("uid5"));
        put5.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid1"), Bytes.toBytes("uid1"));
        put5.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uid4"), Bytes.toBytes("uid4"));
        // 放入List<Put>中
        ArrayList<Put> puts = new ArrayList<Put>();
        puts.add(put1);
        puts.add(put2);
        puts.add(put3);
        puts.add(put4);
        puts.add(put5);
        // 执行插入
        System.out.println("inserting into the table");
        friends.put(puts);
        friends.close();
        System.out.println("insert completed !");


    }

    public void destory(){
        if(admin != null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if(conn != null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
